package pro.cherish.kafka.controller;

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.stereotype.Controller;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RequestParam;
import org.springframework.web.bind.annotation.ResponseBody;
import pro.cherish.kafka.KafkaProducer;
import pro.cherish.kafka.entity.User;


@Controller
@RequestMapping("/api/kafka/")
public class KafkaController {
	@Autowired
	private KafkaTemplate<String, Object> kafkaTemplate;
	@Autowired
	private KafkaProducer kafkaProducer;

	@GetMapping("send")
	@ResponseBody
	public boolean send(@RequestParam String message) {
		try {
			kafkaTemplate.send("testTopic", message);
			System.out.println("消息发送成功...");
		} catch (Exception e) {
			e.printStackTrace();
		}
		return true;
	}

	@GetMapping("sendUser")
	@ResponseBody
	public boolean sendUser(@RequestParam String username, @RequestParam Integer age) {
		try {
			User user = new User();
			user.setName(username);
			user.setAge(age);
			kafkaProducer.sendUserMessage(user);
			System.out.println("发送用户成功");
		} catch (Exception e) {
			e.printStackTrace();
		}
		return true;
	}

	@GetMapping("test")
	@ResponseBody
	public String test() {
		System.out.println("hello world!");
		return "ok";
	}
}